package spring.cloud.tasks.executor.name_space.executor;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.state.ConnectionState;
import spring.cloud.tasks.executor.utils.LogUtils;

import static spring.cloud.tasks.common.Threads.sleep;

@Slf4j
@Setter
@Getter
public abstract class ExecutorConnectionReConnectListener implements org.apache.curator.framework.state.ConnectionStateListener {

    private String executorId;
    private volatile boolean connected = false;
    private volatile boolean closed = false;

    public ExecutorConnectionReConnectListener(String executorId) {
        this.executorId = executorId;
    }

    public void close() {
        this.closed = true;
    }

    @Override
    public void stateChanged(final CuratorFramework curatorFramework, final ConnectionState connectionState) {
        if (closed) {
            return;
        }
        final String id = curatorFramework.toString();
        if (ConnectionState.SUSPENDED == connectionState) {
            connected = false;
            //
            LogUtils.warn(log, "", "The executor {} found zk is SUSPENDED, client is {}", executorId, id);
            final long sessionId = getSessionId(curatorFramework);
            if (!closed) {
                Runnable runnable = new Runnable() {
                    @Override
                    public void run() {
                        do {
                            sleep();
                            if (closed) {
                                break;
                            }
                            long $sessionId = getSessionId(curatorFramework);
                            if (sessionId != $sessionId) {
                                LogUtils.warn(log, ExecutorConnectionReConnectListener.class.getCanonicalName(), "The executor {} is going to restart for zk lost, client is {}", executorId, id);
                                onLost();
                                break;
                            }
                        } while (!closed && !connected);
                    }
                };
                Thread thread = new Thread(runnable);
                thread.setDaemon(true);
                thread.start();
            }
        } else if (ConnectionState.RECONNECTED == connectionState) {
            LogUtils.warn(log, "", "The executor {} found zk is RECONNECTED, client is {}", executorId, id);
            connected = true;
        }
    }

    private long getSessionId(CuratorFramework curatorFramework) {
        long sessionId;
        try {
            sessionId = curatorFramework.getZookeeperClient().getZooKeeper().getSessionId();
        } catch (Exception e) {
            return -1;
        }
        return sessionId;
    }

    public abstract void onLost();


}